package demo.designPattern.observer;

import demo.java.lang.ThreadDemo;
import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.BackpressureStrategy;
import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.core.FlowableEmitter;
import io.reactivex.rxjava3.core.FlowableOnSubscribe;
import io.reactivex.rxjava3.core.FlowableSubscriber;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableEmitter;
import io.reactivex.rxjava3.core.ObservableOnSubscribe;
import io.reactivex.rxjava3.functions.Action;
import io.reactivex.rxjava3.functions.Consumer;
import io.reactivex.rxjava3.schedulers.Schedulers;
import org.junit.Test;
import org.reactivestreams.Subscription;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.List;

/**
 * RxJava – Reactive Extensions for the JVM –
 * a library for composing asynchronous and event-based programs using observable sequences for the Java VM.
 * <p>
 * ReactiveX 是一个用于异步编程的库，旨在使处理流和事件更简单、更直观。
 * 它使用观察者模式（Observer Pattern）来处理异步数据流，并通过函数式编程的方式处理这些数据流的各个阶段。
 * 观察者模式：定义了一对多的依赖关系，当一个对象状态改变时，所有依赖于它的对象都会得到通知并自动更新。
 * 推送（Push）与拉取（Pull）：在 RxJava 中，Observable 通常以推送的方式提供数据，而传统的集合使用拉取的方式。
 * <p>
 * RxJava 的核心概念。存在一个可观察对象(Observable)，观察者(Observer)订阅(Subscribe)它，当数据就绪时，之前定义的机制就会分发数据给一直处于等待状态的观察者哨兵。
 * 有时我们也将Observer叫做Subscriber、Watcher、Reactor。这个模型通常被称作Reactor模式。
 * <li>Observable: 一个数据流的生产者，可以发射（emit）数据和事件。它可以是热（Hot）或冷（Cold）的，热Observable会在数据发生时立即发布通知，而冷Observable则是懒惰的，只有在订阅后才会开始发布数据。
 * <li>Observer: 监听Observable并响应其发射的事件的对象。Observable 发射的数据可以被一个或多个观察者接收。
 * <li>Subscriber: 实现了Observer接口的对象，负责接收Observable发射的事件，并对这些事件作出相应。
 * <li>Subject: 既是Observable又是Observer。它可以用于多播（multicasting）操作，允许多个订阅者接收同一数据流。
 * <li>Operator: 用于转换、过滤和组合Observable的数据流的函数。RxJava提供了丰富的操作符，例如map、filter、flatMap、merge等，可以操作和组合数据流。
 * <li>Schedulers: 用于控制异步操作的线程，RxJava提供了多个类型的Scheduler，例如Schedulers.io()用于I/O操作，Schedulers.computation()用于计算操作等。
 * <p>
 * RxJava 的优点
 * <li>简化异步编程：通过使用观察者和响应式编程，可以更容易地处理复杂的异步操作和事件。
 * <li>组合性：允许开发者通过操作符组合多个数据源和操作链，使得代码结构更清晰。
 * <li>错误处理：提供了集中化的错误处理机制，使得处理异常更加方便。
 * <li>线程控制：灵活地管理线程，减少了在多线程环境中的复杂性。
 */
public class RxJavaDemo {

    private static final Logger logger = LoggerFactory.getLogger(RxJavaDemo.class);

    @Test
    public void testObservable() {
        // 1. 创建被观察者
        Observable<String> observable1 = Observable.just("Hello", "World");

        // 从Iterable集合创建 Observable
        List<String> items = Arrays.asList("Item1", "Item2", "Item3");
        Observable<String> observable3 = Observable.fromIterable(items);

        // 使用 observable.create
        Observable<String> observable2 = Observable.create(
                new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(ObservableEmitter<String> emitter) {
                        emitter.onNext("Hello");
                        emitter.onNext("World");
                        throw new NullPointerException("Throw a Exception...");
                    }
                });


        // 2. 创建观察者
        Consumer<String> onNext = new Consumer<String>() {
            @Override
            public void accept(String s) throws Throwable {
                System.out.println(s);
            }
        };

        Consumer<Throwable> onError = new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Throwable {
                System.err.println("Error: " + throwable.getMessage());
            }
        };

        Action onComplete = new Action() {
            @Override
            public void run() throws Throwable {
                System.out.println("Completed");
            }
        };

        // 3. 订阅事件
        observable2.map(String::toUpperCase)      // 使用map操作符转换数据
                .subscribe(onNext, onError, onComplete); // 订阅并打印结果
    }
    /**
     * 定义一个订阅者
     */
    public static FlowableSubscriber<String> subscriber = new FlowableSubscriber<String>() {
        @Override
        public void onSubscribe(Subscription s) {
            s.request(Long.MAX_VALUE); // 请求无限量数据
        }

        @Override
        public void onNext(String item) {
            logger.info("Received: {}", item); // 处理接收到的数据项
        }

        @Override
        public void onError(Throwable t) {
            logger.error("Error: {}", t.getMessage());// 处理错误
        }

        @Override
        public void onComplete() {
            logger.info("Completed"); // 完成处理
        }
    };

    /**
     * Flowable 用于表示一个可观察的数据流。
     * Flowable 实现了 Reactive Streams 规范，特别适合处理可背压（Backpressure）的场景，即当事件生成的速度超过消费者处理的速度时，能够有效地控制数据的流量。
     * <p>
     * 主要特性和功能
     * <li>背压支持：Flowable 旨在处理大量数据流时的背压问题。背压机制允许消费者控制从生产者那里接收的事件数量，以防止因消费速度慢而导致的内存溢出。
     * <li>异步与事件驱动：Flowable 支持异步事件处理，即可以在不同的线程中生成和处理事件。这使得在复杂应用中可以保持高效性能和响应能力。
     * <li>丰富的操作符：Flowable 提供了大量的操作符，例如 map、filter、flatMap、merge、zip 等，用于对数据流进行变换、组合、处理和错误管理等。
     * <li>多种创建方式：包括：
     * 从已有的数据源（如 Iterable、Callable、Future 等）创建。
     * 使用 create() 方法自行定义事件的生成。
     * 使用 interval()、range() 等创建定期生成数据的流。
     */
    @Test
    public void testFlowable() {
        Flowable.just("Hello world").subscribe(System.out::println);

        FlowableOnSubscribe<String> source = new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull FlowableEmitter<String> emitter) throws Throwable {

            }
        };
        Flowable<String> flowable = Flowable.create(source, BackpressureStrategy.BUFFER);
    }

    /**
     * 背压策略（BackpressureStrategy） 是指当生产者生成数据的速度快于消费者处理数据的速度时，如何控制数据流的策略。它帮助系统避免因过载而导致崩溃或性能下降。
     *
     * <h1>背压的必要性</h1>
     * 在异步和事件驱动的系统中，生产者（例如，数据源、网络请求等）可能会生成数据的速度远快于消费者（例如，用户界面、数据库写操作等）处理这些数据的速度。如果不采取措施进行控制，可能会导致以下问题：
     *
     * <li>内存溢出：数据在缓冲区中积压，消耗大量内存。
     * <li>系统崩溃：当内存占用达到极限时，系统可能会崩溃或变得无响应。
     * <li>延迟增加：处理的数据变得延迟，导致用户体验不佳。
     * <p><br>
     * 在 RxJava 中，BackpressureStrategy 是一个枚举类，定义了一些常用的背压策略。
     * <li>ERROR：默认行为。生产者在无法提供更多数据时会抛出 MissingBackpressureException。适用于对性能要求极高但不希望丢失数据的场景。
     * <li>DROP：如果缓存区满，丢弃末尾的项。适用于不需要每一条数据都被处理的场景。
     * <li>LATEST：如果缓存区满，丢弃最早的项，只保留最新的一项。这适用于关注“最新”数据而不是历史数据的场景。
     * <li>BUFFER：使用缓冲区存储所有项，直到消费者可以处理。可导致内存消耗增加，但不会丢失数据。
     * <li>MISSING（在 RxJava3 中）：表示生产者不处理背压，而是将责任交给消费者。这意味着生产者将无限制地发送数据，消费者需要自行管理。
     */
    @Test
    public void testBackpressureStrategy() {
        Flowable<Integer> flowable = Flowable.range(1, 1000) // 生成1到1000的整数
                .onBackpressureDrop() // 使用丢弃策略
                .observeOn(Schedulers.io()); // 在 IO 线程上观察

        flowable.subscribe(
                item -> {
                    logger.info("Processed: " + item);
                    try {
                        Thread.sleep(10); // 模拟处理延迟
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                },
                Throwable::printStackTrace,
                () -> System.out.println("Complete")
        );

        // 保持主线程存活
        ThreadDemo.safeSleep(10_000);
    }

    @Test
    public void testFlowableOnSubscribe() {
        FlowableOnSubscribe<String> source = new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(FlowableEmitter<String> emitter) throws Exception {
                // 发射数据
                for (int i = 0; i < 10; i++) {
                    if (emitter.isCancelled()) {
                        return; // 如果被取消则结束
                    }
                    emitter.onNext(i + ""); // 发送数据项
                }
                emitter.onComplete(); // 完成
            }
        };
        // 创建 Flowable
        Flowable<String> flowable = Flowable.create(source, BackpressureStrategy.BUFFER);  // 指定背压策略

        // 订阅 Flowable
        flowable.subscribe(subscriber);
    }
}
